package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.arithmetic.common.Cn6EngineDataThresholdInfo;
import com.navinfo.tripanalysis.offline.service.LoadCn6ThresholdInfoService;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.math.BigDecimal;
import java.util.*;

public class LoadCn6ThresholdDataInfoServiceImpl implements LoadCn6ThresholdInfoService {
    private String dbName;

    private String tbName;

    private String selectHql;

    public String getDbName() {
        return dbName;
    }

    public void setDbName(String dbName) {
        this.dbName = dbName;
    }

    public String getTbName() {
        return tbName;
    }

    public void setTbName(String tbName) {
        this.tbName = tbName;
    }

    public String getSelectHql() {
        return selectHql;
    }

    public void setSelectHql(String selectHql) {
        this.selectHql = selectHql;
    }

    @Override
    public JavaPairRDD<String, Cn6EngineDataThresholdInfo> getCn6ThresholdInfoRdd(SparkSession spark, JavaSparkContext jsc) {

        String infoHql = String.format(this.selectHql, this.dbName, this.tbName);
        Dataset<Row> dataSet = spark.sql(infoHql);

        JavaPairRDD<String, Cn6EngineDataThresholdInfo> cn6ThresholdInfoRDD = dataSet.toJavaRDD().mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, String, Cn6EngineDataThresholdInfo>) rowIterator -> {
            List<Tuple2<String, Cn6EngineDataThresholdInfo>> cn6ThresholdInfoList = new ArrayList<>();
            while (rowIterator.hasNext()) {
                Row row = rowIterator.next();
                Cn6EngineDataThresholdInfo cn6ThresholdInfo = new Cn6EngineDataThresholdInfo();
                cn6ThresholdInfo.setEngineType(row.getString(0));
                cn6ThresholdInfo.setSpeedThresholdHigh(new BigDecimal(row.getString(1)));
                cn6ThresholdInfo.setSpeedThresholdLow(new BigDecimal(row.getString(2)));
                cn6ThresholdInfo.setBarometricPressureThresholdHigh(new BigDecimal(row.getString(3)));
                cn6ThresholdInfo.setBarometricPressureThresholdLow(new BigDecimal(row.getString(4)));
                cn6ThresholdInfo.setEngineOutputTorqueThresholdHigh(new BigDecimal(row.getString(5)));
                cn6ThresholdInfo.setEngineOutputTorqueThresholdLow(new BigDecimal(row.getString(6)));
                cn6ThresholdInfo.setFrictionTorqueThresholdHigh(new BigDecimal(row.getString(7)));
                cn6ThresholdInfo.setFrictionTorqueThresholdLow(new BigDecimal(row.getString(8)));
                cn6ThresholdInfo.setRotationThresholdHigh(new BigDecimal(row.getString(9)));
                cn6ThresholdInfo.setRotationThresholdLow(new BigDecimal(row.getString(10)));
                cn6ThresholdInfo.setFuelConsumptionRateThresholdHigh(new BigDecimal(row.getString(11)));
                cn6ThresholdInfo.setFuelConsumptionRateThresholdLow(new BigDecimal(row.getString(12)));
                cn6ThresholdInfo.setScrUpstreamNOxSensorOutputThresholdHigh(new BigDecimal(row.getString(13)));
                cn6ThresholdInfo.setScrUpstreamNOxSensorOutputThresholdLow(new BigDecimal(row.getString(14)));
                cn6ThresholdInfo.setScrDownstreamNOxSensorOutputThresholdHigh(new BigDecimal(row.getString(15)));
                cn6ThresholdInfo.setScrDownstreamNOxSensorOutputThresholdLow(new BigDecimal(row.getString(16)));
                cn6ThresholdInfo.setReagentAllowanceThresholdHigh(new BigDecimal(row.getString(17)));
                cn6ThresholdInfo.setReagentAllowanceThresholdLow(new BigDecimal(row.getString(18)));
                cn6ThresholdInfo.setAirInflowThresholdHigh(new BigDecimal(row.getString(19)));
                cn6ThresholdInfo.setAirInflowThresholdLow(new BigDecimal(row.getString(20)));
                cn6ThresholdInfo.setScrInletTemperatureThresholdHigh(new BigDecimal(row.getString(21)));
                cn6ThresholdInfo.setScrInletTemperatureThresholdLow(new BigDecimal(row.getString(22)));
                cn6ThresholdInfo.setScrOutletTemperatureThresholdHigh(new BigDecimal(row.getString(23)));
                cn6ThresholdInfo.setScrOutletTemperatureThresholdLow(new BigDecimal(row.getString(24)));
                cn6ThresholdInfo.setDpfDropoutVoltageThresholdHigh(new BigDecimal(row.getString(25)));
                cn6ThresholdInfo.setDpfDropoutVoltageThresholdLow(new BigDecimal(row.getString(26)));
                cn6ThresholdInfo.setEngineCoolantTemperatureThresholdHigh(new BigDecimal(row.getString(27)));
                cn6ThresholdInfo.setEngineCoolantTemperatureThresholdLow(new BigDecimal(row.getString(28)));
                cn6ThresholdInfo.setTankLevelThresholdHigh(new BigDecimal(row.getString(29)));
                cn6ThresholdInfo.setTankLevelThresholdLow(new BigDecimal(row.getString(30)));
                cn6ThresholdInfo.setTemperatureOutputThresholdHigh(new BigDecimal(row.getString(31)));
                cn6ThresholdInfo.setTemperatureOutputThresholdLow(new BigDecimal(row.getString(32)));
                cn6ThresholdInfo.setTernaryUpO2OutputThresholdHigh(new BigDecimal(row.getString(33)));
                cn6ThresholdInfo.setTernaryUpO2OutputThresholdLow(new BigDecimal(row.getString(34)));
                cn6ThresholdInfo.setTernaryDownO2OutputThresholdHigh(new BigDecimal(row.getString(35)));
                cn6ThresholdInfo.setTernaryDownO2OutputThresholdLow(new BigDecimal(row.getString(36)));
                cn6ThresholdInfoList.add(new Tuple2(cn6ThresholdInfo.getEngineType(), cn6ThresholdInfo));
            }
            return cn6ThresholdInfoList.iterator();
        });
        return cn6ThresholdInfoRDD;

    }
}
